Концепция распределенной обработки данных и технологии удаленной обработки данных

Содержание лекции

• Понятие и архитектура распределенной системы
• Протоколы и программная реализация удаленного вызова процедур
• Модель RPC: контракт, прокси, сериализация, ошибки
• Объектно-ориентированные вызовы удаленных методов
• Архитектура решений, основанных на Web Services
• Протоколы и стандарты
• Публикация и развертывание служб
• Практические примеры на C++ и Qt

Понятие распределенной системы

Определение

Распределенная система — это совокупность независимых компьютеров, представляемых пользователю как единая координированная система.

Основные характеристики

• Автономность компонентов — каждый узел может работать независимо
• Согласованность — единое представление данных
• Масштабируемость — возможность добавления новых узлов
• Отказоустойчивость — продолжение работы при сбоях
• Прозрачность — скрытие распределенности от пользователя

Архитектурные стили распределенных систем

Клиент-серверная архитектура

• Тонкий клиент — минимальная логика на клиенте
• Толстый клиент — значительная часть логики на клиенте
• Многоуровневая архитектура — presentation, business, data layers

Одноранговая (P2P) архитектура

• Все узлы равноправны
• Отсутствует централизованный сервер
• Высокая отказоустойчивость

Удаленный вызов процедур (RPC)

Концепция RPC

Позволяет программе вызывать процедуры на удаленном компьютере так, как если бы они были локальными.

Главная идея

• Клиент работает с локальной функцией или объектом-прокси
• Прокси преобразует вызов в сетевое сообщение
• Сервер принимает сообщение и вызывает реальную процедуру
• Результат или ошибка возвращаются клиенту как ответ

Зачем нужен RPC

Механизм RPC-вызова

Контракт RPC

Что описывает контракт

• Имена доступных методов
• Типы и порядок параметров
• Тип возвращаемого значения
• Возможные ошибки
• Версию интерфейса

Зачем нужен контракт

• Клиент и сервер одинаково понимают формат сообщений
• Ошибки несовместимости обнаруживаются раньше
• Можно генерировать клиентские и серверные заглушки
• Упрощается документирование и тестирование API

Чем RPC отличается от REST

RPC

• Ориентирован на действия и методы
• Пример: calculator.add(5, 3)
• URL часто один: /rpc или /jsonrpc
• Метод операции задается в теле запроса

REST

• Ориентирован на ресурсы и их состояния
• Пример: GET /users/10, POST /orders
• Операция выражается HTTP-методом
• Ресурс выражается URI

Семантика удаленного вызова

Ошибки в RPC

Категории ошибок

• Ошибка транспорта: нет соединения, таймаут, DNS, TLS
• Ошибка протокола: некорректный JSON/XML, неверная версия
• Ошибка контракта: неизвестный метод, неверные параметры
• Ошибка приложения: деление на ноль, нет прав, объект не найден
• Ошибка инфраструктуры: перегрузка, отказ БД, недоступность зависимости

Что должен делать клиент

• Различать сетевые и прикладные ошибки
• Ограничивать время ожидания ответа
• Не повторять небезопасные операции без защиты
• Логировать id запроса и причину сбоя

Синхронный и асинхронный RPC

Ограничения прозрачности RPC

Почему удаленный вызов не равен локальному

• Сетевая задержка на порядки выше вызова функции в памяти
• Сервер может быть недоступен
• Сообщение может потеряться или прийти повторно
• Данные нужно сериализовать и копировать
• Версии клиента и сервера могут отличаться

Вывод

RPC удобно имитирует локальный вызов, но проектировать его нужно как сетевое взаимодействие с отказами, задержками и ограниченными гарантиями.

Пример клиентского прокси RPC

// Интерфейс удаленной процедуры
class CalculatorInterface {
public:
    virtual int add(int a, int b) = 0;
    virtual int multiply(int a, int b) = 0;
    virtual ~CalculatorInterface() = default;
};

// Клиентский прокси
class CalculatorProxy : public CalculatorInterface {
private:
    QTcpSocket* socket;
    
public:
    int add(int a, int b) override {
        // Сериализация запроса
        QJsonObject request;
        request["method"] = "add";
        request["params"] = QJsonArray{a, b};
        request["id"] = generateId();
        
        // Отправка запроса
        sendRequest(request);
        
        // Получение ответа
        QJsonObject response = receiveResponse();
        return response["result"].toInt();
    }
};

Протокол XML-RPC

Структура запроса

<?xml version="1.0"?>
<methodCall>
    <methodName>calculator.add</methodName>
    <params>
        <param><value><int>5</int></value></param>
        <param><value><int>3</int></value></param>
    </params>
</methodCall>

Структура ответа

<?xml version="1.0"?>
<methodResponse>
    <params>
        <param><value><int>8</int></value></param>
    </params>
</methodResponse>

Реализация XML-RPC клиента на Qt

#include <QtNetwork>
#include <QXmlStreamReader>
#include <QXmlStreamWriter>

class XmlRpcClient : public QObject {
    Q_OBJECT
    
private:
    QNetworkAccessManager* manager;
    QString serverUrl;
    
public:
    XmlRpcClient(const QString& url, QObject* parent = nullptr) 
        : QObject(parent), serverUrl(url) {
        manager = new QNetworkAccessManager(this);
    }
    
    void callMethod(const QString& methodName, 
                   const QVariantList& params) {
        // Создание XML запроса
        QByteArray requestData = createXmlRequest(methodName, params);
        
        // Отправка POST запроса
        QNetworkRequest request(QUrl(serverUrl));
        request.setHeader(QNetworkRequest::ContentTypeHeader, 
                         "text/xml");
        
        QNetworkReply* reply = manager->post(request, requestData);
        
        connect(reply, &QNetworkReply::finished, [=]() {
            if (reply->error() == QNetworkReply::NoError) {
                QVariant result = parseXmlResponse(reply->readAll());
                emit methodCalled(result);
            } else {
                emit errorOccurred(reply->errorString());
            }
            reply->deleteLater();
        });
    }
    
private:
    QByteArray createXmlRequest(const QString& method, 
                               const QVariantList& params);
    QVariant parseXmlResponse(const QByteArray& response);
    
signals:
    void methodCalled(const QVariant& result);
    void errorOccurred(const QString& error);
};

Реализация XML-RPC сервера на Qt

#include <QTcpServer>
#include <QTcpSocket>
#include <QXmlStreamReader>

class XmlRpcServer : public QTcpServer {
    Q_OBJECT
    
private:
    QMap<QString, std::function<QVariant(QVariantList)>> methods;
    
public:
    XmlRpcServer(QObject* parent = nullptr) : QTcpServer(parent) {
        // Регистрация методов
        registerMethod("add", [](QVariantList params) {
            return params[0].toInt() + params[1].toInt();
        });
        
        registerMethod("multiply", [](QVariantList params) {
            return params[0].toInt() * params[1].toInt();
        });
    }
    
    void registerMethod(const QString& name, 
                       std::function<QVariant(QVariantList)> method) {
        methods[name] = method;
    }
    
protected:
    void incomingConnection(qintptr socketDescriptor) override {
        QTcpSocket* socket = new QTcpSocket(this);
        socket->setSocketDescriptor(socketDescriptor);
        
        connect(socket, &QTcpSocket::readyRead, [=]() {
            QByteArray request = socket->readAll();
            QByteArray response = processRequest(request);
            socket->write(response);
            socket->disconnectFromHost();
        });
    }
    
private:
    QByteArray processRequest(const QByteArray& request);
};

JSON-RPC протокол

Преимущества JSON-RPC

• Компактность — меньший размер сообщений по сравнению с XML
• Производительность — быстрее парсится
• Читаемость — легче воспринимается человеком
• Поддержка типов — хорошая поддержка в Qt

Пример JSON-RPC запроса

{
    "jsonrpc": "2.0",
    "method": "calculator.divide",
    "params": [10, 2],
    "id": 1
}

Пример JSON-RPC ответа

{
    "jsonrpc": "2.0",
    "result": 5,
    "id": 1
}

Реализация JSON-RPC на Qt

#include <QJsonDocument>
#include <QJsonObject>
#include <QJsonArray>

class JsonRpcClient : public QObject {
    Q_OBJECT
    
private:
    QNetworkAccessManager* manager;
    QString serverUrl;
    int requestId;
    
public:
    JsonRpcClient(const QString& url, QObject* parent = nullptr)
        : QObject(parent), serverUrl(url), requestId(0) {
        manager = new QNetworkAccessManager(this);
    }
    
    void callMethod(const QString& method, const QJsonArray& params) {
        // Формирование запроса
        QJsonObject request;
        request["jsonrpc"] = "2.0";
        request["method"] = method;
        request["params"] = params;
        request["id"] = ++requestId;
        
        QJsonDocument doc(request);
        QByteArray data = doc.toJson();
        
        // Отправка запроса
        QNetworkRequest netRequest(QUrl(serverUrl));
        netRequest.setHeader(QNetworkRequest::ContentTypeHeader, 
                            "application/json");
        
        QNetworkReply* reply = manager->post(netRequest, data);
        
        connect(reply, &QNetworkReply::finished, [=]() {
            if (reply->error() == QNetworkReply::NoError) {
                QJsonDocument responseDoc = 
                    QJsonDocument::fromJson(reply->readAll());
                QJsonObject response = responseDoc.object();
                
                if (response.contains("result")) {
                    emit methodSuccess(response["result"]);
                } else if (response.contains("error")) {
                    emit methodError(response["error"].toObject());
                }
            }
            reply->deleteLater();
        });
    }
    
signals:
    void methodSuccess(const QJsonValue& result);
    void methodError(const QJsonObject& error);
};

Современные RPC-фреймворки

Объектно-ориентированные вызовы удаленных методов

Концепция удаленных объектов

• Удаленные объекты представляются локальными прокси
• Прозрачный доступ к методам удаленных объектов
• Автоматическая сериализация/десериализация параметров
• Состояние объекта хранится на сервере или синхронизируется с клиентом
• Клиент работает не с самим объектом, а с его удаленной ссылкой

Пример удаленного объекта

// Интерфейс удаленного объекта
class IRemoteCalculator : public QObject {
    Q_OBJECT
    
public:
    virtual ~IRemoteCalculator() = default;
    
public slots:
    virtual int add(int a, int b) = 0;
    virtual int subtract(int a, int b) = 0;
    virtual double divide(double a, double b) = 0;
};

// Клиентская реализация
class RemoteCalculatorClient : public IRemoteCalculator {
    Q_OBJECT
    
private:
    QTcpSocket* socket;
    
public:
    RemoteCalculatorClient(QTcpSocket* sock, QObject* parent = nullptr)
        : IRemoteCalculator(parent), socket(sock) {
        connect(socket, &QTcpSocket::readyRead, 
                this, &RemoteCalculatorClient::onReadyRead);
    }
    
    int add(int a, int b) override {
        return invokeRemoteMethod("add", QVariantList{a, b}).toInt();
    }
    
private:
    QVariant invokeRemoteMethod(const QString& method, 
                               const QVariantList& args);
    void onReadyRead();
};

Система удаленных вызовов на Qt

// Базовый класс для удаленных объектов
class RemoteObject : public QObject {
    Q_OBJECT
    
protected:
    QString objectName;
    
public:
    RemoteObject(const QString& name, QObject* parent = nullptr)
        : QObject(parent), objectName(name) {}
    
    virtual QJsonObject toJson() const {
        QJsonObject obj;
        obj["objectName"] = objectName;
        return obj;
    }
    
    virtual void fromJson(const QJsonObject& json) {
        objectName = json["objectName"].toString();
    }
};

// Удаленный калькулятор
class RemoteCalculator : public RemoteObject {
    Q_OBJECT
    
public:
    RemoteCalculator(const QString& name, QObject* parent = nullptr)
        : RemoteObject(name, parent) {}
    
public slots:
    QJsonValue add(const QJsonArray& params) {
        int a = params[0].toInt();
        int b = params[1].toInt();
        return QJsonValue(a + b);
    }
    
    QJsonValue multiply(const QJsonArray& params) {
        int a = params[0].toInt();
        int b = params[1].toInt();
        return QJsonValue(a * b);
    }
};

Архитектура Web Services

SOAP (Simple Object Access Protocol)

• Протокол на основе XML
• Строгая типизация
• Поддержка стандартов безопасности
• WSDL (Web Services Description Language)

REST (Representational State Transfer)

• Простота и легкость
• Использование стандартных HTTP методов
• Stateless архитектура
• Поддержка различных форматов данных

// REST клиент на Qt
class RestClient : public QObject {
    Q_OBJECT
    
private:
    QNetworkAccessManager* manager;
    QString baseUrl;
    
public:
    RestClient(const QString& base, QObject* parent = nullptr)
        : QObject(parent), baseUrl(base) {
        manager = new QNetworkAccessManager(this);
    }
    
    void get(const QString& resource) {
        QNetworkRequest request(QUrl(baseUrl + resource));
        QNetworkReply* reply = manager->get(request);
        
        connect(reply, &QNetworkReply::finished, [=]() {
            handleResponse(reply);
        });
    }
    
    void post(const QString& resource, const QJsonObject& data) {
        QNetworkRequest request(QUrl(baseUrl + resource));
        request.setHeader(QNetworkRequest::ContentTypeHeader, 
                         "application/json");
        
        QJsonDocument doc(data);
        QNetworkReply* reply = manager->post(request, doc.toJson());
        
        connect(reply, &QNetworkReply::finished, [=]() {
            handleResponse(reply);
        });
    }
};

RESTful сервис на Qt

#include <QTcpServer>
#include <QJsonDocument>
#include <QRegularExpression>

class RestServer : public QTcpServer {
    Q_OBJECT
    
private:
    struct Route {
        QString method;
        QRegularExpression pattern;
        std::function<QJsonObject(const QJsonObject&)> handler;
    };
    
    QList<Route> routes;
    
public:
    RestServer(QObject* parent = nullptr) : QTcpServer(parent) {
        // Регистрация маршрутов
        registerRoute("GET", QRegularExpression("^/api/users$"), 
                     [](const QJsonObject& params) {
            QJsonObject response;
            QJsonArray users;
            
            QJsonObject user1;
            user1["id"] = 1;
            user1["name"] = "John";
            users.append(user1);
            
            response["users"] = users;
            return response;
        });
        
        registerRoute("POST", QRegularExpression("^/api/users$"), 
                     [](const QJsonObject& params) {
            QJsonObject response;
            response["id"] = 123;
            response["status"] = "created";
            return response;
        });
    }
    
    void registerRoute(const QString& method, 
                      const QRegularExpression& pattern,
                      std::function<QJsonObject(const QJsonObject&)> handler) {
        routes.append({method, pattern, handler});
    }
    
protected:
    void incomingConnection(qintptr socketDescriptor) override {
        QTcpSocket* socket = new QTcpSocket(this);
        socket->setSocketDescriptor(socketDescriptor);
        
        connect(socket, &QTcpSocket::readyRead, [=]() {
            QByteArray request = socket->readAll();
            QByteArray response = processHttpRequest(request);
            socket->write(response);
            socket->disconnectFromHost();
        });
    }
    
private:
    QByteArray processHttpRequest(const QByteArray& request);
};

Протоколы и стандарты

HTTP/HTTPS

• GET — получение данных
• POST — создание ресурсов
• PUT — обновление ресурсов
• DELETE — удаление ресурсов
• PATCH — частичное обновление

Форматы данных

• JSON — JavaScript Object Notation
• XML — eXtensible Markup Language
• Protocol Buffers — бинарный формат Google
• MessagePack — эффективный бинарный формат

// Поддержка Protocol Buffers в Qt
class ProtobufSerializer {
public:
    static QByteArray serializeUser(const QJsonObject& user) {
        // Упрощенная реализация
        QByteArray data;
        QDataStream stream(&data, QIODevice::WriteOnly);
        
        stream << user["id"].toInt();
        stream << user["name"].toString();
        stream << user["email"].toString();
        
        return data;
    }
    
    static QJsonObject deserializeUser(const QByteArray& data) {
        QJsonObject user;
        QDataStream stream(data);
        
        int id;
        QString name, email;
        stream >> id >> name >> email;
        
        user["id"] = id;
        user["name"] = name;
        user["email"] = email;
        
        return user;
    }
};

Безопасность в распределенных системах

Аутентификация и авторизация

class SecurityManager : public QObject {
    Q_OBJECT
    
private:
    QString secretKey;
    int tokenExpiration;
    
public:
    SecurityManager(const QString& key, QObject* parent = nullptr)
        : QObject(parent), secretKey(key), tokenExpiration(3600) {}
    
    QString generateToken(const QString& userId) {
        QJsonObject payload;
        payload["userId"] = userId;
        payload["exp"] = QDateTime::currentDateTime()
                        .addSecs(tokenExpiration).toSecsSinceEpoch();
        
        // Простая JWT-подобная реализация
        QString header = QString::fromUtf8(
            QJsonDocument({{"alg", "HS256"}, {"typ", "JWT"}}).toJson(
                QJsonDocument::Compact).toBase64());
        
        QString payloadStr = QString::fromUtf8(
            QJsonDocument(payload).toJson(
                QJsonDocument::Compact).toBase64());
        
        QString signature = QCryptographicHash::hash(
            (header + "." + payloadStr).toUtf8(),
            QCryptographicHash::Sha256).toHex();
        
        return header + "." + payloadStr + "." + signature;
    }
    
    bool validateToken(const QString& token) {
        QStringList parts = token.split('.');
        if (parts.size() != 3) return false;
        
        // Проверка подписи
        QString expectedSignature = QCryptographicHash::hash(
            (parts[0] + "." + parts[1]).toUtf8(),
            QCryptographicHash::Sha256).toHex();
        
        if (parts[2] != expectedSignature) return false;
        
        // Проверка срока действия
        QJsonDocument payloadDoc = QJsonDocument::fromJson(
            QByteArray::fromBase64(parts[1].toUtf8()));
        
        qint64 exp = payloadDoc.object()["exp"].toDouble();
        return QDateTime::currentDateTime().toSecsSinceEpoch() < exp;
    }
};

Шифрование данных

class EncryptionManager {
public:
    static QByteArray encryptAES(const QByteArray& data, 
                                const QByteArray& key) {
        // Упрощенная реализация AES шифрования
        QByteArray encrypted;
        for (int i = 0; i < data.size(); ++i) {
            encrypted.append(data[i] ^ key[i % key.size()]);
        }
        return encrypted;
    }
    
    static QByteArray decryptAES(const QByteArray& encrypted, 
                                const QByteArray& key) {
        // AES дешифрование
        return encryptAES(encrypted, key); // XOR шифрование симметрично
    }
    
    static QSslSocket* createSecureConnection(
        const QString& host, quint16 port) {
        
        QSslSocket* socket = new QSslSocket();
        
        // Настройка SSL/TLS
        connect(socket, &QSslSocket::encrypted, []() {
            qDebug() << "Соединение зашифровано";
        });
        
        connect(socket, &QSslSocket::sslErrors, 
                [](const QList<QSslError>& errors) {
            for (const QSslError& error : errors) {
                qDebug() << "SSL ошибка:" << error.errorString();
            }
        });
        
        socket->connectToHostEncrypted(host, port);
        return socket;
    }
};

Публикация и развертывание служб

class ServiceConfiguration {
public:
    struct ServiceInfo {
        QString name;
        QString version;
        QString description;
        QString endpoint;
        quint16 port;
        bool sslEnabled;
    };
    
    static ServiceInfo loadConfiguration(const QString& configFile) {
        ServiceInfo info;
        
        QFile file(configFile);
        if (!file.open(QIODevice::ReadOnly)) {
            qWarning() << "Не удалось открыть файл конфигурации";
            return info;
        }
        
        QJsonDocument doc = QJsonDocument::fromJson(file.readAll());
        QJsonObject config = doc.object();
        
        info.name = config["serviceName"].toString();
        info.version = config["version"].toString("1.0.0");
        info.description = config["description"].toString();
        info.endpoint = config["endpoint"].toString();
        info.port = config["port"].toInt(8080);
        info.sslEnabled = config["sslEnabled"].toBool(false);
        
        return info;
    }
    
    static void saveConfiguration(const ServiceInfo& info, 
                                 const QString& configFile) {
        QJsonObject config;
        config["serviceName"] = info.name;
        config["version"] = info.version;
        config["description"] = info.description;
        config["endpoint"] = info.endpoint;
        config["port"] = info.port;
        config["sslEnabled"] = info.sslEnabled;
        
        QJsonDocument doc(config);
        
        QFile file(configFile);
        if (file.open(QIODevice::WriteOnly)) {
            file.write(doc.toJson());
        }
    }
};

Система развертывания

#include <QProcess>
#include <QFileSystemWatcher>

class ServiceDeployment : public QObject {
    Q_OBJECT
    
private:
    QString servicePath;
    QProcess* serviceProcess;
    QFileSystemWatcher* watcher;
    
public:
    ServiceDeployment(const QString& path, QObject* parent = nullptr)
        : QObject(parent), servicePath(path) {
        
        serviceProcess = new QProcess(this);
        watcher = new QFileSystemWatcher(this);
        
        // Наблюдение за изменениями в каталоге службы
        watcher->addPath(servicePath);
        
        connect(watcher, &QFileSystemWatcher::directoryChanged,
                this, &ServiceDeployment::onServiceChanged);
        
        connect(serviceProcess, QOverload<int, QProcess::ExitStatus>::of(
                &QProcess::finished),
                this, &ServiceDeployment::onServiceFinished);
    }
    
    void deploy() {
        // Остановка текущей службы
        stopService();
        
        // Копирование новых файлов
        copyServiceFiles();
        
        // Запуск службы
        startService();
    }
    
    void startService() {
        QString program = servicePath + "/service";
        QStringList arguments;
        arguments << "--config" << servicePath + "/config.json";
        
        serviceProcess->start(program, arguments);
        
        if (serviceProcess->waitForStarted()) {
            emit serviceStarted();
        } else {
            emit errorOccurred("Не удалось запустить службу");
        }
    }
    
    void stopService() {
        if (serviceProcess->state() != QProcess::NotRunning) {
            serviceProcess->terminate();
            if (!serviceProcess->waitForFinished(5000)) {
                serviceProcess->kill();
            }
        }
    }
    
signals:
    void serviceStarted();
    void serviceStopped();
    void errorOccurred(const QString& error);
    
private slots:
    void onServiceChanged(const QString& path) {
        qDebug() << "Служба изменена, перезапуск...";
        deploy();
    }
    
    void onServiceFinished(int exitCode, QProcess::ExitStatus exitStatus) {
        emit serviceStopped();
        if (exitCode != 0) {
            emit errorOccurred(QString("Служба завершилась с кодом %1")
                             .arg(exitCode));
        }
    }
};

Мониторинг и логирование

#include <QFile>
#include <QTextStream>
#include <QDateTime>

class ServiceLogger : public QObject {
    Q_OBJECT
    
private:
    QFile logFile;
    QtMsgType logLevel;
    
public:
    ServiceLogger(const QString& logPath, QtMsgType level = QtInfoMsg,
                 QObject* parent = nullptr)
        : QObject(parent), logLevel(level) {
        
        logFile.setFileName(logPath);
        if (logFile.open(QIODevice::WriteOnly | QIODevice::Append)) {
            qInstallMessageHandler(messageHandler);
        }
    }
    
    static void messageHandler(QtMsgType type, 
                              const QMessageLogContext& context,
                              const QString& msg) {
        static ServiceLogger* logger = 
            qobject_cast<ServiceLogger*>(qApp->property("logger").value<QObject*>());
        
        if (!logger) return;
        
        QString timestamp = QDateTime::currentDateTime()
                           .toString("yyyy-MM-dd hh:mm:ss.zzz");
        
        QString logLevel;
        switch (type) {
            case QtDebugMsg: logLevel = "DEBUG"; break;
            case QtInfoMsg: logLevel = "INFO"; break;
            case QtWarningMsg: logLevel = "WARNING"; break;
            case QtCriticalMsg: logLevel = "CRITICAL"; break;
            case QtFatalMsg: logLevel = "FATAL"; break;
        }
        
        QString logMessage = QString("[%1] %2: %3 (%4:%5)\n")
                           .arg(timestamp)
                           .arg(logLevel)
                           .arg(msg)
                           .arg(context.file)
                           .arg(context.line);
        
        QTextStream stream(&logger->logFile);
        stream << logMessage;
        stream.flush();
        
        // Вывод в консоль
        fprintf(stderr, "%s", logMessage.toLocal8Bit().constData());
        
        if (type == QtFatalMsg) {
            abort();
        }
    }
};

class ServiceMonitor : public QObject {
    Q_OBJECT
    
private:
    QTimer* statsTimer;
    QJsonObject currentStats;
    
public:
    ServiceMonitor(QObject* parent = nullptr) : QObject(parent) {
        statsTimer = new QTimer(this);
        statsTimer->setInterval(60000); // 1 минута
        
        connect(statsTimer, &QTimer::timeout,
                this, &ServiceMonitor::collectStatistics);
        
        statsTimer->start();
    }
    
    void updateMetric(const QString& name, const QVariant& value) {
        currentStats[name] = QJsonValue::fromVariant(value);
    }
    
    QJsonObject getStatistics() const {
        return currentStats;
    }
    
signals:
    void statisticsUpdated(const QJsonObject& stats);
    
private slots:
    void collectStatistics() {
        // Сбор системной информации
        currentStats["timestamp"] = QDateTime::currentDateTime().toString(Qt::ISODate);
        currentStats["memoryUsage"] = getMemoryUsage();
        currentStats["cpuUsage"] = getCpuUsage();
        currentStats["activeConnections"] = getActiveConnections();
        
        emit statisticsUpdated(currentStats);
    }
    
private:
    qint64 getMemoryUsage() {
        // Упрощенное получение информации о памяти
        return 0; // Реальная реализация зависит от платформы
    }
    
    double getCpuUsage() {
        // Упрощенное получение информации о CPU
        return 0.0;
    }
    
    int getActiveConnections() {
        // Получение количества активных соединений
        return 0;
    }
};

Примеры интеграции

// Пример полноценного распределенного калькулятора
class DistributedCalculator : public QObject {
    Q_OBJECT
    
private:
    JsonRpcClient* client;
    ServiceMonitor* monitor;
    
public:
    DistributedCalculator(const QString& serverUrl, 
                         QObject* parent = nullptr)
        : QObject(parent) {
        
        client = new JsonRpcClient(serverUrl, this);
        monitor = new ServiceMonitor(this);
        
        connect(client, &JsonRpcClient::methodSuccess,
                this, &DistributedCalculator::onCalculationSuccess);
        
        connect(client, &JsonRpcClient::methodError,
                this, &DistributedCalculator::onCalculationError);
        
        connect(monitor, &ServiceMonitor::statisticsUpdated,
                this, &DistributedCalculator::onStatisticsUpdated);
    }
    
    void calculate(const QString& operation, 
                  const QJsonArray& params) {
        monitor->updateMetric("lastOperation", operation);
        monitor->updateMetric("operationsCount", 
                             monitor->getStatistics()["operationsCount"]
                             .toInt() + 1);
        
        client->callMethod(operation, params);
    }
    
    void performComplexCalculation() {
        // Пример цепочки удаленных вызовов
        QJsonArray params1{10, 5};
        client->callMethod("add", params1);
        
        QJsonArray params2{8, 3};
        client->callMethod("multiply", params2);
        
        QJsonArray params3{100, 4};
        client->callMethod("divide", params3);
    }
    
signals:
    void calculationResult(const QString& operation, 
                          double result);
    void calculationError(const QString& error);
    void systemStatistics(const QJsonObject& stats);
    
private slots:
    void onCalculationSuccess(const QJsonValue& result) {
        double value = result.toDouble();
        emit calculationResult("operation", value);
    }
    
    void onCalculationError(const QJsonObject& error) {
        QString errorMsg = error["message"].toString();
        emit calculationError(errorMsg);
    }
    
    void onStatisticsUpdated(const QJsonObject& stats) {
        emit systemStatistics(stats);
    }
};

// Пример использования
int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);
    
    DistributedCalculator calculator("http://localhost:8080/jsonrpc");
    
    QObject::connect(&calculator, &DistributedCalculator::calculationResult,
                    [](const QString& op, double result) {
        qDebug() << "Результат:" << result;
    });
    
    QObject::connect(&calculator, &DistributedCalculator::calculationError,
                    [](const QString& error) {
        qDebug() << "Ошибка:" << error;
    });
    
    // Простые вычисления
    calculator.calculate("add", QJsonArray{15, 25});
    calculator.calculate("multiply", QJsonArray{7, 8});
    
    // Сложные вычисления
    calculator.performComplexCalculation();
    
    return app.exec();
}

Масштабируемость и производительность

class LoadBalancer : public QObject {
    Q_OBJECT
    
private:
    struct ServerInfo {
        QString url;
        int load;
        bool available;
        QDateTime lastCheck;
    };
    
    QList<ServerInfo> servers;
    QTimer* healthCheckTimer;
    
public:
    LoadBalancer(QObject* parent = nullptr) : QObject(parent) {
        healthCheckTimer = new QTimer(this);
        healthCheckTimer->setInterval(30000); // 30 секунд
        
        connect(healthCheckTimer, &QTimer::timeout,
                this, &LoadBalancer::performHealthCheck);
        
        // Добавление серверов
        addServer("http://server1:8080");
        addServer("http://server2:8080");
        addServer("http://server3:8080");
        
        healthCheckTimer->start();
    }
    
    void addServer(const QString& url) {
        ServerInfo info;
        info.url = url;
        info.load = 0;
        info.available = true;
        info.lastCheck = QDateTime::currentDateTime();
        
        servers.append(info);
    }
    
    QString selectServer() {
        // Алгоритм наименьшей нагрузки
        ServerInfo* bestServer = nullptr;
        int minLoad = INT_MAX;
        
        for (ServerInfo& server : servers) {
            if (server.available && server.load < minLoad) {
                minLoad = server.load;
                bestServer = &server;
            }
        }
        
        if (bestServer) {
            bestServer->load++;
            return bestServer->url;
        }
        
        return QString(); // Нет доступных серверов
    }
    
private slots:
    void performHealthCheck() {
        for (ServerInfo& server : servers) {
            // Проверка доступности сервера
            QNetworkAccessManager* manager = new QNetworkAccessManager();
            QNetworkRequest request(QUrl(server.url + "/health"));
            
            QNetworkReply* reply = manager->get(request);
            
            connect(reply, &QNetworkReply::finished, [=]() {
                bool wasAvailable = server.available;
                server.available = (reply->error() == QNetworkReply::NoError);
                server.lastCheck = QDateTime::currentDateTime();
                
                if (wasAvailable != server.available) {
                    emit serverAvailabilityChanged(server.url, 
                                                  server.available);
                }
                
                reply->deleteLater();
                manager->deleteLater();
            });
        }
    }
    
signals:
    void serverAvailabilityChanged(const QString& url, bool available);
};

class ScalableService : public QObject {
    Q_OBJECT
    
private:
    LoadBalancer* loadBalancer;
    QThreadPool* threadPool;
    
public:
    ScalableService(QObject* parent = nullptr) : QObject(parent) {
        loadBalancer = new LoadBalancer(this);
        threadPool = new QThreadPool(this);
        threadPool->setMaxThreadCount(10);
    }
    
    void processRequest(const QJsonObject& request) {
        // Выбор сервера
        QString serverUrl = loadBalancer->selectServer();
        if (serverUrl.isEmpty()) {
            emit errorOccurred("Нет доступных серверов");
            return;
        }
        
        // Асинхронная обработка в пуле потоков
        QtConcurrent::run(threadPool, [=]() {
            JsonRpcClient client(serverUrl);
            
            QEventLoop loop;
            QJsonObject result;
            bool success = false;
            
            connect(&client, &JsonRpcClient::methodSuccess,
                    [&](const QJsonValue& response) {
                result["data"] = response;
                success = true;
                loop.quit();
            });
            
            connect(&client, &JsonRpcClient::methodError,
                    [&](const QJsonObject& error) {
                result["error"] = error;
                loop.quit();
            });
            
            client.callMethod("process", QJsonArray{request});
            loop.exec();
            
            emit requestProcessed(result, success);
        });
    }
    
signals:
    void requestProcessed(const QJsonObject& result, bool success);
    void errorOccurred(const QString& error);
};

Заключение

Ключевые концепции распределенных систем

• Прозрачность — скрытие сложности распределенной природы
• Масштабируемость — возможность роста системы
• Отказоустойчивость — продолжение работы при сбоях
• Согласованность — поддержание целостности данных
• Безопасность — защита данных и коммуникаций

Современные технологии

• gRPC — высокопроизводительный RPC от Google
• Apache Thrift — кросс-платформенный RPC фреймворк
• Message Queue — асинхронные системы обмена сообщениями
• Microservices — архитектурный стиль распределенных приложений

Инструменты Qt для распределенных систем

• Qt Network — низкоуровневые сетевые операции
• Qt WebSockets — двунаправленная связь в реальном времени
• Qt HTTP — высокоуровневые HTTP операции
• Qt JSON — сериализация данных
• Qt Concurrent — параллельные вычисления

Вопросы для самопроверки

1. Чем отличается RPC от REST архитектуры?
2. Какие преимущества и недостатки у JSON-RPC по сравнению с XML-RPC?
3. Как реализовать безопасность в распределенных системах?
4. Что такое прозрачность в контексте распределенных систем?
5. Какие механизмы обеспечивают отказоустойчивость?
6. Как реализовать балансировку нагрузки в распределенной системе?
7. Какие протоколы используются для удаленного вызова процедур?
8. Как обеспечить согласованность данных в распределенной системе?
9. Чем отличается клиент-серверная архитектура от P2P?
10. Какие современные технологии используются для создания распределенных приложений?

Резюме

Основные достижения

• Разобрали архитектуру распределенных систем
• Изучили протоколы удаленного вызова процедур
• Реализовали примеры на C++ и Qt
• Рассмотрели вопросы безопасности и масштабируемости
• Изучили методы публикации и развертывания служб

Дальнейшее изучение

• gRPC и Protocol Buffers
• Apache Thrift
• Message Queue системы (RabbitMQ, Apache Kafka)
• Микросервисная архитектура
• Контейнеризация и оркестрация (Docker, Kubernetes)
• Облачные технологии (AWS, Azure, Google Cloud)

Qt предоставляет мощные инструменты для создания распределенных приложений, объединяя простоту разработки с высокой производительностью.